import os
import sys
import clr
import uuid
import json
import copy
import time
import uuid
import redis
import caches
import models
import pickle
import logging
import traceback
import threading
import itertools
import datetime
import multiprocessing
from typing import Any, Dict
from typing import List
from typing import Union
from typing import Optional
from sqlalchemy import asc
from sqlalchemy import and_
from sqlalchemy import desc
from sqlalchemy import text
from datetime import timedelta
from models.course.course import Course
from models.course.detailed import Detailed
from models.course.sample import Sample
from models.course.serial import Serial
from models.facility.awaits import Awaits
from models.facility.facility import Facility
from models.logs.stationlog import StationLog
from models.order.tasks import Tasks
from models.owner.owner import Owner
from models.owner.entering import Entering
from models.product.aircraft import Aircraft
from models.product.routes import Routes
from models.facility.andon import Andon
from models.product.craft import Craft
from models.facility.status import Status
from models.facility.shields import Shields
from public.local import datetime_type, nest
from public.path import get_warns_path

base_dirs = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
dll_dirs: str = base_dirs + os.sep + "library" + os.sep + "Library.dll"

exec("clr.AddReference(dll_dirs)")
exec("from Library import *")


def digital_vbl(t_parm: Union[int, float, None]) -> bool:
    return type(t_parm) in [int, float]


def digits_cut(t_parm: Union[int, float, list], length: int = 4) -> Union[int, float, list]:
    if type(t_parm) in [int, float]:
        return round(t_parm, length)
    for f_itr, f_value in enumerate(t_parm):
        t_parm[f_itr] = round(f_value, length)
    return t_parm


class WorkWS0204(threading.Thread):
    _instance = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super(WorkWS0204, cls).__new__(cls)
        return cls._instance

    def __init__(self, ft_id: int = 19, protocol: str = "NanoSerialOverTcp", host: str = "192.168.1.40", port: int = 8501):
        super(WorkWS0204, self).__init__()
        self.daemon = True
        self.interval = 0.8
        self.ft_id: int = ft_id
        self.protocol: str = protocol
        self.connect = eval("{}('{}', {})".format(self.protocol, host, port))
        self.IsNet: bool = False

        self.status: Union[dict, None] = None
        self.alarms: dict = {"tp": None, "status": None}
        self.awaits: Union[dict, None] = None
        self.shields: Union[dict, None] = None

        # ###################################################################################
        self.uuid4: Union[str, None] = None
        self.cache = redis.Redis(connection_pool=caches.POOL)
        self.session = models.SessionFactory()
        self.SNS: List[dict] = []
        self.heartbeat: int = 0

    def __initialize(self):
        for f_itr in range(8):
            self.SNS.append({
            "SN": "", "work": 0, "cu_id": 0, "last": 0, "at_id": 0, "tk_id": 0,
            "status": 0, "start": time.strftime("%Y-%m-%d %H:%M:%S"), "uuid4": None
        })
        db_facility: Union[Facility, None] = self.session.query(Facility).filter(Facility.id == self.ft_id).first()
        if db_facility is None:
            raise ValueError("NOT FIND BY FT_ID")
        self.uuid4 = str(db_facility.uuid4)

        if db_facility.alarms is not None:
            self.alarms["tp"] = pickle.loads(db_facility.alarms)
        if  self.alarms["tp"] is None:
            with open(get_warns_path(), "rb") as file:
                self.alarms["tp"] = pickle.loads(file.read())

    def processing_status(self, t_status: int):
        # noinspection PyBroadException
        try:
            if self.IsNet is True:
                t_option = {
                    1: "未运行",
                    2: "初始化中",
                    4: "初始化完成",
                    8: "自动运行中",
                    16: "设备报警中",
                }
                r_message: Union[str, None] = t_option.get(t_status)
                if r_message is None:
                    r_message = "PLC未给正确状态码"

                if self.status is None:
                    self.status = {
                        "start": time.time(),
                        "desc": r_message,
                        "session": uuid.uuid4(),
                        "trigger": time.time()
                    }
                    op_user: Union[str, None] = None
                    db_op_user: Union[Entering, None] = self.session.query(Entering).filter(
                                Entering.facility_id == self.ft_id
                            ).first()
                    if db_op_user is not None:
                        op_user = db_op_user.option
                    self.session.add(Status(
                        time=datetime_type(),
                        facility_id=self.ft_id,
                        code=t_status,
                        message=r_message,
                        account=op_user,
                        expenditure=0,
                        session=self.status.get("session")
                    ))
                    self.session.commit()

                if self.status.get("desc") != r_message:
                    db_status: Union[Status, None] = self.session.query(Status).filter(
                                Status.session == self.status.get("session")
                            ).order_by(desc(Status.id)).first()
                    if db_status is not None and db_status.finish is None:
                        db_status.finish = datetime_type()
                        db_status.expenditure = (datetime_type() - db_status.time).seconds

                    op_user: Union[str, None] = None
                    db_op_user: Union[Entering, None] = self.session.query(Entering).filter(
                                Entering.facility_id == self.ft_id
                            ).first()
                    self.status = {
                        "start": time.time(),
                        "desc": r_message,
                        "session": uuid.uuid4(),
                        "trigger": time.time()
                    }
                    if db_op_user is not None:
                        op_user = db_op_user.option
                    self.session.add(Status(
                        time=datetime_type(),
                        facility_id=self.ft_id,
                        code=t_status,
                        message=r_message,
                        account=op_user,
                        expenditure=0,
                        session=self.status.get("session")
                    ))
                    self.session.commit()

                if time.time() - self.status.get("start") >= 32:
                    db_status: Union[Status, None] = self.session.query(Status).filter(
                                Status.session == self.status.get("session")
                            ).order_by(desc(Status.id)).first()
                    if db_status:
                        db_status.expenditure = (datetime_type() - db_status.time).seconds
                    self.status["start"] = time.time()
                    self.session.commit()

                if time.time() - self.status.get("trigger") > 7200:
                    db_status: Union[Status, None] = self.session.query(Status).filter(
                                Status.session == self.status.get("session")
                            ).order_by(desc(Status.id)).first()
                    if db_status is not None:
                        db_status.finish = datetime_type()
                        db_status.expenditure = (datetime_type() - db_status.time).seconds
                        self.session.add(Status(
                            time=datetime_type(),
                            facility_id=self.ft_id,
                            code=db_status.code,
                            message=db_status.message,
                            account=db_status.account,
                            expenditure=0,
                            session=self.status.get("session")
                        ))
                    self.status["start"] = time.time()
                    self.status["trigger"] = time.time()
                    self.session.commit()
        except Exception as e:
            self.status = None
            logging.error(msg=traceback.print_exc())

    def processing_alarms(self, t_status: int):
        ignore_array: List[str] = [
            "气压异常报警",
            "安全门打开报警"
        ]
        t_option: Dict[int, str] = {
            1: "未运行",
            2: "初始化中",
            4: "初始化完成",
            8: "自动运行中",
            16: "设备报警中",
        }
        r_message: Union[str, None] = t_option.get(t_status)
        if r_message is None:
            r_message = "PLC未给正确状态码"
        redis_status_key: str = "status-{}".format(self.uuid4)
        # noinspection PyBroadException
        try:
            if self.IsNet is True and self.alarms.get("tp") is not None:
                status_options: List[dict] = self.alarms["tp"]["Alarm"]
                run_status: Union[dict, None] = self.alarms.get("status")

                if t_status == 16 and run_status is None:
                    for itr_dict in status_options:
                        error_key_kind = itr_dict.get("Key")
                        plc_addr: str = str(itr_dict.get("Io"))
                        addr_code: int = self.connect.ReadShort(plc_addr)
                        addr_alarm: Dict[int, str] = itr_dict.get("Option")
                        if addr_code in list(addr_alarm.keys()) and addr_alarm.get(addr_code) not in ignore_array:
                            error_key_context = addr_alarm.get(addr_code)
                            self.alarms["status"] = {
                                    "start": time.time(),
                                    "session": uuid.uuid4(),
                                    "trigger": time.time(),
                                    "error_context": error_key_context,
                                }
                            op_user: Union[str, None] = None
                            db_op_user: Union[Entering, None] = self.session.query(Entering).filter(
                                        Entering.facility_id == self.ft_id
                                    ).first()
                            if db_op_user is not None:
                                op_user = db_op_user.option
                            self.session.add(Andon(
                                time=datetime_type(),
                                facility_id=self.ft_id,
                                kind=error_key_kind,
                                message=error_key_context,
                                account=op_user,
                                expenditure=0,
                                session=self.alarms["status"].get("session")
                            ))
                            self.session.commit()
                            break

                if self.alarms["status"] is not None and (time.time() - self.alarms["status"]["start"]) >= 16:
                    db_andon: Union[Andon, None] = self.session.query(Andon).filter(
                                Andon.session == self.alarms["status"]["session"]
                            ).order_by(desc(Andon.id)).first()
                    if db_andon is not None:
                        db_andon.expenditure = (datetime_type() - db_andon.time).seconds
                    self.alarms["status"]["start"] = time.time()
                    self.session.commit()

                if self.alarms["status"] and (time.time() - self.alarms["status"]["trigger"]) >= 7200:
                    db_andon: Union[Andon, None] = self.session.query(Andon).filter(
                                Andon.session == self.alarms["status"]["session"]
                            ).order_by(desc(Andon.id)).first()
                    if db_andon:
                        db_andon.finish = datetime_type()
                        db_andon.expenditure = (datetime_type() - db_andon.time).seconds
                        self.session.add(Andon(
                            time=datetime_type(),
                            facility_id=self.ft_id,
                            kind=db_andon.kind,
                            message=db_andon.message,
                            account=db_andon.account,
                            expenditure=0,
                            session=self.alarms["status"].get("session")
                        ))
                        self.alarms["status"]["start"] = time.time
                        self.alarms["status"]["trigger"] = time.time()
                        self.session.commit()

                if run_status is not None and t_status != 16:
                    db_andon: Union[Andon, None] = self.session.query(Andon).filter(
                                Andon.session == self.alarms["status"]["session"]
                            ).order_by(desc(Andon.id)).first()
                    if db_andon:
                        db_andon.finish = datetime_type()
                        db_andon.expenditure = (datetime_type() - db_andon.time).seconds
                    self.alarms["status"] = None
                    self.session.commit()

                self.cache.set(name=redis_status_key, value=t_status, ex=4)
                if self.alarms["status"] is not None:
                    r_message = self.alarms["status"]["error_context"]
                self.cache.set(name=self.uuid4, value=r_message, ex=4)
        except Exception as e:
            self.alarms["status"] = None
            logging.error(msg=traceback.print_exc())

    def processing_awaits(self, t_status: int):
        # noinspection PyBroadException
        try:
            if self.IsNet is True:
                if t_status >= 1 and self.awaits is None:
                    self.awaits = {
                        "start": time.time(),
                        "session": uuid.uuid4(),
                        "trigger": time.time()
                    }
                    op_user: Union[str, None] = None
                    db_op_user: Union[Entering, None] = self.session.query(Entering).filter(
                                Entering.facility_id == self.ft_id
                            ).first()
                    if db_op_user is not None:
                        op_user = db_op_user.option
                    self.session.add(Awaits(
                        time=datetime_type(),
                        facility_id=self.ft_id,
                        account=op_user,
                        expenditure=0,
                        session=self.awaits["session"]
                    ))
                    self.session.commit()

                if t_status > 0 and self.awaits is not None and (time.time() - self.awaits.get("start")) >= 16:
                    db_await: Union[Awaits, None] = self.session.query(Awaits).filter(
                                    Awaits.session == self.awaits["session"]
                                ).order_by(desc(Awaits.id)).first()
                    if db_await and db_await.finish is None:
                        db_await.expenditure = (datetime_type() - db_await.time).seconds
                        self.session.commit()
                    self.awaits["start"] = time.time()

                if t_status > 0 and self.awaits is not None and (time.time() - self.awaits.get("trigger")) >= 3600:
                    db_await: Union[Awaits, None] = self.session.query(Awaits).filter(
                                Awaits.session == self.awaits["session"]
                            ).order_by(desc(Awaits.id)).first()
                    if db_await:
                        db_await.finish = datetime_type()
                        db_await.expenditure = (datetime_type() - db_await.time).seconds
                        self.session.add(Awaits(
                            time=datetime_type(),
                            facility_id=self.ft_id,
                            account=db_await.account,
                            expenditure=0,
                            session=self.awaits["session"]
                        ))
                    self.awaits["start"] = time.time()
                    self.awaits["trigger"] = time.time()
                    self.session.commit()

                if t_status == 0 and self.awaits is not None:
                    db_await: Union[Awaits, None] = self.session.query(Awaits).filter(
                            Awaits.session == self.awaits["session"]
                        ).order_by(desc(Awaits.id)).first()
                    if db_await is not None and db_await.finish is None:
                        db_await.finish = datetime_type()
                        db_await.expenditure = (datetime_type() - db_await.time).seconds
                        self.session.commit()
                    self.awaits = None
                    self.session.commit()
        except Exception as e:
            self.awaits = None
            logging.error(msg=traceback.print_exc())

    def processing_shields(self, t_status: int):
        # noinspection PyBroadException
        try:
            if self.IsNet is True:
                if t_status >= 1 and self.shields is None:
                    self.shields = {
                        "start": time.time(),
                        "trigger": time.time(),
                        "session": uuid.uuid4(),
                    }
                    op_user: Union[str, None] = None
                    db_op_user: Union[Entering, None] = self.session.query(Entering).filter(
                        Entering.facility_id == self.ft_id
                    ).first()
                    if db_op_user:
                        op_user = db_op_user.option
                    self.session.add(Shields(
                        time=datetime_type(),
                        facility_id=self.ft_id,
                        account=op_user,
                        expenditure=0,
                        session=self.shields["session"]
                    ))
                    self.session.commit()

                if t_status > 0 and self.shields is not None and (time.time() - self.shields.get("start")) >= 16:
                    db_shield: Union[Shields, None] = self.session.query(Shields).filter(
                                Shields.session == self.shields["session"]
                            ).order_by(desc(Shields.id)).first()
                    if db_shield is not None:
                        db_shield.expenditure = (datetime_type() - db_shield.time).seconds
                        self.session.commit()
                    self.shields["start"] = time.time()

                if t_status > 0 and self.shields is not None and (time.time() - self.shields.get("trigger")) >= 3600:
                    db_shield: Union[Shields, None] = self.session.query(Shields).filter(
                                Shields.session == self.shields["session"]
                            ).order_by(desc(Shields.id)).first()
                    if db_shield is not None:
                        db_shield.finish = datetime_type()
                        db_shield.expenditure = (datetime_type() - db_shield.time).seconds
                        self.session.add(Shields(
                            time=datetime_type(),
                            facility_id=self.ft_id,
                            account=db_shield.account,
                            expenditure=0,
                            session=self.shields["session"]
                        ))
                    self.shields["start"] = time.time()
                    self.shields["trigger"] = time.time()
                    self.session.commit()

                if t_status == 0 and self.awaits is not None:
                    db_shield: Union[Shields, None] = self.session.query(Shields).filter(
                                Shields.session == self.shields["session"]
                            ).order_by(desc(Shields.id)).first()
                    if db_shield is not None and db_shield.finish is None:
                        db_shield.finish = datetime_type()
                        db_shield.expenditure = (datetime_type() - db_shield.time).seconds
                    self.shields = None
                    self.session.commit()
        except Exception as e:
            self.shields = None
            logging.error(msg=traceback.print_exc())

    def processing_queues(self):
        # noinspection PyBroadException
        try:
            if self.IsNet is True:
                queues_key: str = "queues-{}".format(self.uuid4)
                while True:
                    length: int = self.cache.llen(queues_key)
                    if length > 0:
                        instruct: str = self.cache.lpop(queues_key)
                        instruct: dict = json.loads(instruct)
                        if instruct.get("type") == "Bool":
                            value: bool = True if instruct.get("value") == "True" else False
                            self.connect.WriteBool(instruct.get("address"), value)
                        if instruct.get("type") == "Short":
                            value: int = int(instruct.get("value"))
                            self.connect.WriteShort(instruct.get("address"), value)
                        if instruct.get("type") == "Int":
                            value: int = int(instruct.get("value"))
                            self.connect.WriteInt(instruct.get("address"), value)
                        if instruct.get("type") == "Float":
                            value: float = float(instruct.get("value"))
                            self.connect.WriteFloat(instruct.get("address"), value)
                        if instruct.get("type") == "String":
                            self.connect.WriteString(instruct.get("address"), instruct.get("value"))
                    else:
                        break
        except Exception as e:
            logging.error(msg="QUEUES:= {}".format(traceback.print_exc()))


    def persistence_sample(self, plc_code: int, t_message: Union[dict, None]) -> bool:
        plc_status: bool = True if plc_code == 1 else False
        mes_status: bool = True if plc_code == 1 else False

        if plc_status is True and t_message is not None:
            array_collect: List[dict] = t_message.get("data")
            for f_collect in array_collect:
                if digital_vbl(f_collect.get("value")) and digital_vbl(f_collect.get("floor")):
                    if f_collect.get("value") < f_collect.get("floor"):
                        mes_status = False
                        break
                if digital_vbl(f_collect.get("value")) and digital_vbl(f_collect.get("upper")):
                    if f_collect.get("value") > f_collect.get("upper"):
                        mes_status = False
                        break
        op_user: Entering = self.session.query(Entering).filter(Entering.facility_id == self.ft_id).first()
        db_sample: Sample = Sample(
            time=datetime_type(),
            sequence="样件产品",
            facility_id=self.ft_id,
            message=json.dumps(t_message) if t_message is not None else None,
            finish=datetime_type(),
            option=op_user.option if op_user else None,
            status_mes=mes_status,
            status_plc=plc_status
        )
        self.session.add(db_sample)
        self.session.commit()
        return mes_status


    def permission_main(self, t_index: int, t_sequence: Union[str, None], t_formula: int) -> tuple:
        self.SNS[t_index] = {
            "SN": "", "work": 0, "cu_id": 0, "last": 0, "at_id": 0, "tk_id": 0,
            "status": 0, "start": time.strftime("%Y-%m-%d %H:%M:%S"), "uuid4": uuid.uuid4()
        }
        if t_sequence is None or len(t_sequence) == 0:
            return False, "空条码、禁止加工"
        self.SNS[t_index]["SN"] = t_sequence

        t_task_parm: Union[bool, None] = True
        db_task: Union[Tasks, None] = self.session.query(Tasks).filter(Tasks.status == t_task_parm).first()
        db_course: Union[Course, None] = self.session.query(Course).filter(Course.sequence == t_sequence).first()
        if (db_course is not None) and db_course.mark == 1:
            return False, "已全部加工完成、禁止加工"
        elif (db_course is not None) and db_course.mark == 2 and db_course.repair is None:
            return False, "NG未返工、禁止加工"
        elif (db_course is not None) and db_course.mark == 3:
            return False, "报废产品、禁止加工"

        t_process: Union[str, dict, list, None] = None
        if (db_course is not None) and db_course.mark == 0:
            db_aircraft = self.session.query(Aircraft).filter(Aircraft.id == db_course.aircraft_id).first()
            db_route: Routes = self.session.query(Routes).filter(Routes.id == db_aircraft.route_id).first()
            db_route_json: dict = json.loads(db_route.route)
            t_process: list = db_route_json.get("data")
            self.SNS[t_index]["cu_id"] = db_course.id
            self.SNS[t_index]["tk_id"] = db_course.tasks_id
            self.SNS[t_index]["at_id"] = db_course.aircraft_id
            self.SNS[t_index]["uuid4"] = db_course.session
        elif (db_course is not None) and db_course.mark == 2:
            db_route_json: dict = json.loads(db_course.repair)
            t_process: list = db_route_json.get("data")
            self.SNS[t_index]["cu_id"] = db_course.id
            self.SNS[t_index]["tk_id"] = db_course.tasks_id
            self.SNS[t_index]["at_id"] = db_course.aircraft_id
            self.SNS[t_index]["uuid4"] = db_course.session

        if t_process is None and db_task is None:
            return False, "请先打开对应工单"
        elif t_process is None and db_task is not None:
            db_aircraft: Aircraft = self.session.query(Aircraft).filter(Aircraft.id == db_task.aircraft_id).first()
            db_route: Routes = self.session.query(Routes).filter(Routes.id == db_aircraft.route_id).first()
            db_route_json: dict = json.loads(db_route.route)
            t_process: list = db_route_json.get("data")
            self.SNS[t_index]["tk_id"] = db_task.id
            self.SNS[t_index]["at_id"] = db_task.aircraft_id

        db_formula: Aircraft = self.session.query(Aircraft).filter(Aircraft.id == self.SNS[t_index]["at_id"]).first()
        db_detailed: Union[Detailed, None] = self.session.query(Detailed).filter(and_(
                        Detailed.sequence == t_sequence, Detailed.session == self.SNS[t_index].get("uuid4")
                    )).order_by(desc(Detailed.id)).first()

        if db_detailed is None:
            init_array: list = []
            for first_rts in t_process:
                init_array: list = first_rts.get("ft_ids")
                break
            if self.ft_id not in init_array:
                return False, "工序错误, 禁止加工"
            parm_data: list = db_formula.aircraft.split("-")
            formula_id = int(parm_data[len(parm_data) - 1])
            if t_formula != formula_id:
                return False, "机种配方不一致, 禁止加工"
            self.SNS[t_index]["work"] = 1
            if len(t_process) == 1:
                self.SNS[t_index]["last"] = 1
            return True, "验证工序允许加工"

        fb_work_index: int = -1
        for f_index_nb, first_rts in enumerate(t_process):
            if db_detailed.facility_id in first_rts.get("ft_ids"):
                fb_work_index = f_index_nb

        if self.ft_id not in t_process[fb_work_index + 1].get("ft_ids"):
            return False, "工序错误, 禁止加工"

        parm_data: list = db_formula.aircraft.split("-")
        formula_id = int(parm_data[len(parm_data) - 1])
        if t_formula != formula_id:
            return False, "机种配方不一致, 禁止加工"

        if len(t_process) == (fb_work_index + 1) + 1:
            self.SNS[t_index]["last"] = 1
        self.SNS[t_index]["work"] = 1
        return True, "验证工序允许加工"


    # 生产数据永久保存模块
    def persistence_data(self, t_index: int, plc_code: int, t_message: Union[dict, None]) -> bool:
        if self.SNS[t_index].get("cu_id") == 0:
            self.session.add(Course(
                sequence=self.SNS[t_index].get("SN"), aircraft_id=self.SNS[t_index].get("at_id"),
                tasks_id=self.SNS[t_index].get("tk_id"), start=datetime_type(), mark=0, repair=None,
                session=self.SNS[t_index].get("uuid4")
            ))
            self.session.commit()
            self.session.commit()
            db_t_course: Course = self.session.query(Course).filter(Course.sequence == self.SNS[t_index].get("SN")).first()
            self.SNS[t_index]["cu_id"] = db_t_course.id

        plc_single_bool: bool = True if plc_code == 1 else False
        mes_mark_value: int = 1 if plc_single_bool else 2

        t_first_product: bool = False
        t_backup: Union[str, None] = None
        db_owner: List[Owner] = self.session.query(Owner).order_by(asc(Owner.id)).all()
        for owner in db_owner:
            temp_start = time.strftime("%Y-%m-%d") + " " + owner.start.strftime("%H:%M:%S")
            temp_finish = time.strftime("%Y-%m-%d") + " " + owner.finish.strftime("%H:%M:%S")
            temp_start1 = datetime.datetime.strptime(temp_start, "%Y-%m-%d %H:%M:%S")
            temp_finish1 = datetime.datetime.strptime(temp_finish, "%Y-%m-%d %H:%M:%S")

            if (temp_finish1 > temp_start1) is False:
                temp_finish = nest() + " " + owner.finish.strftime("%H:%M:%S")
                temp_finish1 = datetime.datetime.strptime(temp_finish, "%Y-%m-%d %H:%M:%S")
            if temp_start1 <= datetime_type() <= temp_finish1:
                init_course: List[Detailed] = self.session.query(Detailed).filter(and_(
                        Detailed.time >= temp_start1, Detailed.time < temp_finish1, Detailed.facility_id == self.ft_id
                    )).order_by(asc(Detailed.id)).first()
                if init_course is None:
                    t_first_product = True
                break

        if plc_single_bool is False and t_message is not None:
            array_collect: List[dict] = t_message.get("data")
            for f_collect in array_collect:
                if digital_vbl(f_collect.get("value")) and digital_vbl(f_collect.get("floor")):
                    if f_collect.get("value") < f_collect.get("floor"):
                        t_backup = "下限判定{}不合格".format(f_collect.get("key"))
                        break
                if digital_vbl(f_collect.get("value")) and digital_vbl(f_collect.get("upper")):
                    if f_collect.get("value") > f_collect.get("upper"):
                        t_backup = "上限判定{}不合格".format(f_collect.get("key"))
                        break

        op_user: Entering = self.session.query(Entering).filter(Entering.facility_id == self.ft_id).first()
        t_numbers = self.session.query(Detailed.sequence).filter(and_(
                    Detailed.sequence == self.SNS[t_index].get("SN"),
                    Detailed.facility_id == self.ft_id)
                ).count()
        db_detailed = Detailed(
            time=datetime.datetime.strptime(self.SNS[t_index].get("start"), "%Y-%m-%d %H:%M:%S"),
            sequence=self.SNS[t_index].get("SN"),
            complete=True if self.SNS[t_index].get("last") == 1 else False,
            course_id=self.SNS[t_index].get("cu_id"),
            aircraft_id=self.SNS[t_index].get("at_id"),
            tasks_id=self.SNS[t_index].get("tk_id"),
            facility_id=self.ft_id,
            message=json.dumps(t_message) if t_message is not None else None,
            finish=datetime_type(),
            status=plc_single_bool,
            numbers=t_numbers + 1,
            error=t_backup,
            option=op_user.option if op_user else None,
            mark=mes_mark_value,
            first=t_first_product,
            session=self.SNS[t_index].get("uuid4")
        )
        self.session.add(db_detailed)
        self.SNS[t_index]["status"] = mes_mark_value

        db_cs: Course = self.session.query(Course).filter(Course.sequence == self.SNS[t_index].get("SN")).first()
        if plc_single_bool is False:
            db_cs.mark = 2
            db_cs.repair = None
            db_cs.session = uuid.uuid4()
            db_cs.finish = datetime_type()
            db_task: Tasks = self.session.query(Tasks).filter(Tasks.id == self.SNS[t_index].get("tk_id")).first()
            if db_task is not None:
                db_task.totals = self.session.query(Course.id).filter(and_(Course.mark >= 1, Course.tasks_id == self.SNS[t_index].get("tk_id"))).count()
        if self.SNS[t_index]["last"] == 1 and plc_single_bool is True:
            if db_cs is not None:
                db_cs.mark = 1
                db_cs.finish = datetime_type()
            task_id: int = self.SNS[t_index].get("tk_id")
            db_task: Union[Tasks, None] = self.session.query(Tasks).filter(Tasks.id == task_id).first()
            if db_task is not None:
                db_task.actual = self.session.query(Course.id).filter(and_(Course.mark == 1, Course.tasks_id == task_id)).count()
                db_task.totals = self.session.query(Course.id).filter(and_(Course.mark >= 1, Course.tasks_id == task_id)).count()
                if db_task.actual >= db_task.amount:
                    db_task.status = False
                if db_task.actual >= db_task.amount and db_task.delivery is None:
                    db_task.delivery = datetime_type()
        self.session.commit()
        return True

    def run(self):
        self.__initialize()
        while True:
            start: float = time.perf_counter()
            if self.connect.IsConnect() is True:
                self.IsNet = True
            else:
                self.IsNet = False
            self.session.expire_all()

            status_code: int = self.connect.ReadShort("DM8040")

            # 记录运行状记录模块
            self.processing_status(status_code)
            # 记录报警状记录模块
            self.processing_alarms(status_code)
            # 记录设备待机模块
            self.processing_awaits(0)
            # 记录设备屏蔽MES模块
            self.processing_shields(0)
            # 处理缓存数据写入PLC模块
            self.processing_queues()


            # MES与PLC之间数据交互逻辑代码
            # noinspection PyBroadException
            try:
                if self.IsNet is True:
                    r_parm: dict = {
                        "DM8196": "Short",
                        "DM8060": "Short",
                        "DM8190": "Short",
                        "DM8100": "String",
                        "DM8194": "Short",
                    }
                    data_str_tq_main = self.connect.ReadAll(json.dumps(r_parm))
                    plc_json: dict = json.loads(data_str_tq_main)

                    req_index: int = 0
                    if plc_json.get("DM8190") >= 1 and plc_json.get("DM8196") == 1:
                        time.sleep(0.1)
                        plc_json["DM8100"] = self.connect.ReadString("DM8100")

                        self.session.commit()
                        self.session.expire_all()
                        r_status, r_msg = self.permission_main(req_index, plc_json.get("DM8100"), plc_json.get("DM8060"))

                        r_data_value = 1 if r_status else 2
                        self.connect.WriteShort("DM8191", r_data_value)
                        self.connect.WriteShort("DM8190", 0)
                        self.session.add(StationLog(
                            time=datetime_type(),
                            sequence=plc_json.get("DM8100"),
                            ft_id=self.ft_id,
                            title="请求是否可以加工",
                            address="DM8100",
                            value=str(plc_json.get("DM8190")),
                            feedback=r_msg,
                            datainfo=str(self.SNS[req_index]),
                            dataios=data_str_tq_main
                        ))
                        self.session.commit()

                    req_index: int = 0
                    if plc_json.get("DM8194") >= 1:
                        plc_read_code: int = plc_json.get("DM8194")

                        test_ul_value = list(self.connect.ReadFloatArray("DM8200", 4))
                        test_ul_value = digits_cut(test_ul_value)

                        test_value = list(self.connect.ReadFloatArray("DM8300", 2))
                        test_value = digits_cut(test_value)

                        db_array: List[dict] = [
                            {
                                "key": "位移最大值",
                                "floor": test_ul_value[1],
                                "value": test_value[0],
                                "upper": test_ul_value[0]
                            },
                            {
                                "key": "压力最大值",
                                "floor": test_ul_value[3],
                                "value": test_value[1],
                                "upper": test_ul_value[2]
                            },
                        ]

                        log_title: Union[str, None] = None
                        t_feedback: str = "数据保存失败"
                        save_message: dict = {"data": db_array}
                        if plc_json.get("DM8196") == 1:
                            log_title = "正常交互数据上传信号"

                            if self.SNS[req_index].get("work") == 1 and self.SNS[req_index].get("status") == 0:
                                t_feedback = "正常加工数据保存成功"
                                self.persistence_data(req_index, plc_read_code, save_message)

                            if t_feedback != "正常加工数据保存成功":
                                req_index = 7
                                self.permission_main(req_index, plc_json.get("DM8100"), plc_json.get("DM8060"))
                                if self.SNS[req_index].get("work") == 1 and self.SNS[req_index].get("status") == 0:
                                    t_feedback = "正常加工数据保存成功"
                                    self.persistence_data(req_index, plc_read_code, save_message)

                        if plc_json.get("DM8196") == 2:
                            log_title = "样件数据上传信号"
                            t_feedback = "样件数据保存成功"
                            self.persistence_sample(plc_read_code, save_message)

                        self.session.add(StationLog(
                            time=datetime_type(),
                            sequence=self.SNS[req_index].get("SN"),
                            ft_id=self.ft_id,
                            title=log_title,
                            address="DM8194",
                            value=str(plc_json.get("DM8194")),
                            feedback=t_feedback,
                            datainfo=str(self.SNS),
                            dataios=data_str_tq_main
                        ))
                        self.connect.WriteShort("DM8194", 0)
                        self.session.commit()
            except Exception as e:
                self.session.rollback()
                logging.error(msg="ERROR:= {}".format(traceback.print_exc()))

            # MES 与PLC 心跳频率设置
            # noinspection PyBroadException
            try:
                if self.IsNet is True:
                    plc_addr: str = "DM8400"
                    if self.heartbeat == 0:
                        self.connect.WriteShort(plc_addr, 1)
                        self.heartbeat = 1
                    elif self.heartbeat == 1:
                        self.connect.WriteShort(plc_addr, 0)
                        self.heartbeat = 0
            except Exception as e:
                logging.error(msg="ERROR:= {}".format(traceback.print_exc()))

            self.session.commit()

            cache_ct_key: str = "ct-{}".format(self.uuid4)
            elapsed: float = time.perf_counter() - start
            self.cache.set(cache_ct_key, "{:.5f}".format(elapsed), ex=8)
            time.sleep(max(0.0, self.interval - elapsed))


